package com.nuanshui.framework.schedule.schedule;

import com.google.gson.*;
import com.nuanshui.framework.lang.SysStringUtils;
import com.nuanshui.framework.schedule.conf.ConfUtils;
import com.nuanshui.framework.schedule.conf.ZkConfig;
import com.nuanshui.framework.schedule.zk.Zk;
import com.nuanshui.framework.schedule.zk.entity.ZkData;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.BooleanUtils;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.Type;
import java.nio.charset.Charset;
import java.sql.Timestamp;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.*;

/**
 * zk操作
 */
public class ScheduleDataManager {
    private static final Gson gson = new GsonBuilder().registerTypeAdapter(Timestamp.class, new TimestampTypeAdapter()).setDateFormat("yyyy-MM-dd HH:mm:ss").create();
    private static final Logger LOGGER = LoggerFactory.getLogger("schedule");

    private ZkConfig config = ConfUtils.loadConfig();
    private String pathServer = config.getRootPath() + "/" + config.getServerNode();
    private String pathTask = config.getRootPath() + "/" + config.getTaskNode();
    private String ignoreTaskServerPath = config.getRootPath() + "/" + config.getIgnoreTaskServerNode();
    private Random random = new Random();
    private long zkBaseTime = 0;
    private long localBaseTime = 0;
    private Zk zk;

    private Zk getZk() {
        if (zk == null) {
            zk = new Zk(this.config.getCxnString());
        }
        return zk;
    }

    public ScheduleDataManager() throws Exception {
        getZk().createIfNotExists(this.pathServer);
        String tempPath = getZk().create(config.getRootPath() + "/appTime", null, CreateMode.EPHEMERAL_SEQUENTIAL);
        localBaseTime = System.currentTimeMillis();
        ZkData zkData = getZk().readData(tempPath);
        zkBaseTime = zkData.getStat().getCtime();
        getZk().deleteRecursive(tempPath);
        if (Math.abs(zkBaseTime - localBaseTime) > 2000) {
            LOGGER.error("warning!!! Zookeeper服务器时间与本地时间相差:{}ms", zkBaseTime - localBaseTime);
        }
    }

    /**
     * 注册server
     * @param server server
     * @throws Exception
     */
    public void registerServer(ScheduleServer server) throws Exception {
        if (server.isRegister()) {
            throw new Exception(server.getUuid() + " 被重复注册");
        }
        String realPath;
        //此处必须增加UUID作为唯一性保障
        String id = String.format("%s$%s$", server.getIp(), UUID.randomUUID().toString().replaceAll("-", "").toUpperCase());
        String zkServerPath = pathServer + "/" + id;
        realPath = getZk().create(zkServerPath, null, CreateMode.PERSISTENT_SEQUENTIAL);
        server.setUuid(realPath.substring(realPath.lastIndexOf("/") + 1));
        Timestamp heartBeatTime = new Timestamp(getSystemTime());
        server.setHeartBeatTime(heartBeatTime);
        String serverGson = gson.toJson(server);

        if(MapUtils.isNotEmpty(config.getIgnoreTaskMap())) {
            for (Map.Entry<String, Boolean> e : config.getIgnoreTaskMap().entrySet()) {
                ignoreTask(e.getKey(), server.getUuid(), e.getValue());
            }
        }
        getZk().edit(realPath, serverGson.getBytes(Charset.forName("UTF-8")));
        server.setRegister(true);
    }

    /**
     * 移除server
     * @param server server
     * @throws Exception
     */
    public void removeServer(ScheduleServer server) throws Exception {
        if (!server.isRegister()) {
            throw new Exception(server.getUuid() + " 被重复移除");
        }
        String zkPath = this.pathServer + "/" + server.getUuid();
        if(getZk().exists(zkPath)) {
            getZk().deleteRecursive(zkPath);
        }
        server.setRegister(false);
    }

    /**
     * 发送心跳信息
     */
    public boolean refreshServer(ScheduleServer server) throws Exception {
        Timestamp heartBeatTime = new Timestamp(getSystemTime());
        String zkPath = this.pathServer + "/" + server.getUuid();
        if (!getZk().exists(zkPath)) {
            //数据可能被清除，先清除内存数据后，重新注册数据
            server.setRegister(false);
            return false;
        } else {
            Timestamp oldHeartBeatTime = server.getHeartBeatTime();
            server.setHeartBeatTime(heartBeatTime);
            server.setVersion(server.getVersion() + 1);
            String valueString = gson.toJson(server);
            try {
                getZk().edit(zkPath, valueString.getBytes(Charset.forName("UTF-8")));
            } catch (Exception e) {
                //恢复上次的心跳时间
                server.setHeartBeatTime(oldHeartBeatTime);
                server.setVersion(server.getVersion() - 1);
                throw e;
            }
            return true;
        }
    }

    /**
     * 移除过期的server
     * @throws Exception
     */
    public void clearExpireServer() throws Exception {
        String zkPath = this.pathServer;
        getZk().createIfNotExists(zkPath);
        List<String> serverIds = new ArrayList<String>();
        for (String name : getZk().getChildren(zkPath)) {
            try {
                String nodePath = zkPath + "/" + name;
                ZkData data = getZk().readData(nodePath);
                if (getSystemTime() - data.getStat().getMtime() > config.getServerExpireTime()) {
                    getZk().deleteRecursive(nodePath);
                    serverIds.add(name);
                    LOGGER.debug("ScheduleServer[{}]过期清除", nodePath);
                }
            } catch (Exception e) {
                // 当有多台服务器时，存在并发清理的可能，忽略异常
            }
        }

        if(CollectionUtils.isNotEmpty(serverIds)) {
            getZk().createIfNotExists(ignoreTaskServerPath);
            List<String> tasks = getZk().getChildren(ignoreTaskServerPath);
            if(CollectionUtils.isNotEmpty(tasks)) {
                for (String serverId: serverIds) {
                    for (String taskName: tasks) {
                        removeIgnoreTask(taskName, serverId, false);
                    }
                }
            }
        }

    }

    /**
     * 暂停任务
     * @param taskName 任务名
     * @param serverId serverId
     * @param isAllNode true:全部节点,false:serverId所在节点
     */
    public void ignoreTask(String taskName, String serverId, Boolean isAllNode) {
        if(BooleanUtils.isTrue(isAllNode)) {
            getZk().createIfNotExists(pathServer);
            List<String> servers = getZk().getChildren(pathServer);
            if(CollectionUtils.isNotEmpty(servers)) {
                for (String server: servers) {
                    String ignorePath = ignoreTaskServerPath + "/" + taskName + "/"  + server;
                    getZk().createIfNotExists(ignorePath);
                }
            }
        } else {
            String ignorePath = ignoreTaskServerPath + "/" + taskName + "/"  + serverId;
            getZk().createIfNotExists(ignorePath);
        }
    }

    /**
     * 清理停掉某个节点或者全部节点上的某个任务的限制
     */
    public void removeIgnoreTask(String taskName, String serverId, Boolean isAllNode) {
        String ignoreTaskPath = ignoreTaskServerPath + "/" + taskName;
        if(BooleanUtils.isTrue(isAllNode)) {
            getZk().createIfNotExists(ignoreTaskPath);
            List<String> nodes = getZk().getChildren(ignoreTaskPath);
            if(CollectionUtils.isNotEmpty(nodes)) {
                for (String node : nodes) {
                    removeIgnoreTask(taskName, node, false);
                }
            }
        } else {
            try {
                String ignorePath = ignoreTaskPath + "/"  + serverId;
                getZk().deleteRecursive(ignorePath);
            } catch (Throwable t) {
                // ignore
            }
        }
    }

    /**
     * 获取所有节点
     * @return 所有节点
     */
    public List<String> loadServerNames() {
        String zkPath = this.pathServer;
        if (!getZk().exists(zkPath)) {
            return new ArrayList<String>();
        }
        List<String> serverList = getZk().getChildren(zkPath);
        Collections.sort(serverList, new Comparator<String>() {
            public int compare(String u1, String u2) {
                return u1.substring(u1.lastIndexOf("$") + 1).compareTo(u2.substring(u2.lastIndexOf("$") + 1));
            }
        });
        return serverList;
    }

    /**
     * 获取所有任务
     * @return 所有任务
     */
    public List<String> loadTaskNames() {
        List<String> tasks = new ArrayList<String>();
        String zkPath = this.pathTask;
        try {
            getZk().createIfNotExists(zkPath);
            tasks = getZk().getChildren(zkPath);
        } catch (Exception e) {
            LOGGER.warn(e.getMessage(), e);
        }
        return tasks;
    }

    /**
     * 获取执行任务的server
     * @param taskName taskName
     * @return 执行任务的server
     */
    public List<String> loadTaskServers(String taskName) {
        List<String> tasks = new ArrayList<String>();
        String zkPath = this.pathTask + "/" + taskName;
        try {
            getZk().createIfNotExists(zkPath);
            tasks = getZk().getChildren(zkPath);
        } catch (Exception e) {
            LOGGER.warn(e.getMessage(), e);
        }
        return tasks;
    }

    /**
     * 获取任务的忽略节点(就是这个任务在哪些节点不执行)
     * @param taskName taskName
     * @return 任务的忽略节点
     */
    public List<String> loadIgnoreTaskServers(String taskName) {
        List<String> ignoreServers = new ArrayList<String>();
        try {
            String ignorePath = ignoreTaskServerPath + "/" + taskName;
            getZk().createIfNotExists(ignorePath);
            ignoreServers = getZk().getChildren(ignorePath);
        } catch (Exception e) {
            LOGGER.warn(e.getMessage(), e);
        }
        return ignoreServers;
    }

    @SuppressWarnings("unchecked")
	public void assignTask(String currentUuid, List<String> taskServerList) throws Exception {
        if (!isLeader(currentUuid, taskServerList)) {
            LOGGER.debug("不是负责任务分配的Leader,直接返回,uuid:{}",currentUuid);
            return;
        }
        LOGGER.debug("开始重新分配任务...,uuid:{}", currentUuid);
        if (CollectionUtils.isEmpty(taskServerList)) {
            //在服务器动态调整的时候，可能出现服务器列表为空的情况
            LOGGER.debug("服务器列表为空!uuid:{}",currentUuid);
            return;
        }
        String zkPath = this.pathTask;
        getZk().createIfNotExists(zkPath);
        List<String> taskList = getZk().getChildren(zkPath);

        if (CollectionUtils.isEmpty(taskList)) {
            LOGGER.debug("没有集群任务,uuid:{}",currentUuid);
            return;
        }
        LOGGER.debug(new Gson().toJson(taskList));
        for (String taskName : taskList) { // 遍历task
            String taskPath = zkPath + "/" + taskName;
            List<String> taskServerIds = loadTaskServers(taskName); // 正在执行任务的server
            LOGGER.debug("{}:{}",taskName, new Gson().toJson(taskServerIds));
            List<String> ignoreServers = loadIgnoreTaskServers(taskName); // 不执行此任务的server
			List<String> pTaskServerList = (List<String>) CollectionUtils.subtract(taskServerList, ignoreServers); // 可以执行此任务的server
            if(CollectionUtils.isEmpty(pTaskServerList)) {
                LOGGER.debug("任务[{}]没有可执行节点",taskName);
                continue;
            }
            if (CollectionUtils.isEmpty(taskServerIds)) {
                assignServer2Task(pTaskServerList, taskPath);
            } else {
                if(config.isRunInAll()) { // run in all
                    @SuppressWarnings("rawtypes")
					Collection remain = CollectionUtils.subtract(pTaskServerList, taskServerIds); // 没有被分配到执行权限的server
					Collection<String> remove = CollectionUtils.subtract(taskServerIds, pTaskServerList); // 以前分配过执行权限的server，但现在不在可执行此任务列表里的server
                    if(CollectionUtils.isNotEmpty(remain)) {
                        assignServer2Task(new ArrayList<String>(remain), taskPath);
                    }
                    if(CollectionUtils.isNotEmpty(remove)) {
                        for (String s :remove) {
                            String taskNode = taskPath + "/" + s;
                            getZk().deleteRecursive(taskNode);
                            LOGGER.info("task node[{}]过期清除", taskNode);
                        }
                    }
                } else { // run in one
                    boolean hasAssignSuccess = false;
                    // 在可执行此任务的server集合里，找出
                    for (String serverId : taskServerIds) {
                        if (pTaskServerList.contains(serverId)) {
                            hasAssignSuccess = true;
                            continue;
                        }
                        // 如果在taskServerIds里，
                        getZk().deleteRecursive(taskPath + "/" + serverId);
                    }
                    if (!hasAssignSuccess) {
                        assignServer2Task(pTaskServerList, taskPath);
                    }
                }
            }
        }

    }

    private void assignServer2Task(List<String> taskServerList, String taskPath) throws Exception {
        if(config.isRunInAll()) { // run in all
            for (String serverId : taskServerList) {
                getZk().createIfNotExists(taskPath + "/" + serverId);
            }
            LOGGER.debug("Assign server [{}] to task [{}]", SysStringUtils.join(taskServerList, ","), taskPath);
        } else { // run in one
            int index = random.nextInt(taskServerList.size());
            String serverId = taskServerList.get(index);
            getZk().createPath(taskPath + "/" + serverId);
            LOGGER.debug("Assign server [{}] to task [{}]", serverId, taskPath);
        }
    }

    public boolean isLeader(String uuid, List<String> serverList) {
        return SysStringUtils.equals(uuid, getLeader(serverList));
    }

    public String getLeader(List<String> serverList) {
        if (CollectionUtils.isEmpty(serverList)) {
            return SysStringUtils.EMPTY;
        }
        long no = Long.MAX_VALUE;
        long tmpNo;
        String leader = null;
        for (String server : serverList) {
            tmpNo = Long.parseLong(server.substring(server.lastIndexOf("$") + 1));
            if (no > tmpNo) {
                no = tmpNo;
                leader = server;
            }
        }
        return leader;
    }

    public long getSystemTime() {
        return this.zkBaseTime + (System.currentTimeMillis() - this.localBaseTime);
    }

    public boolean isOwner(String name, String uuid) throws Exception {
        //查看集群中是否注册当前任务，如果没有就自动注册
        String zkPath = this.pathTask + "/" + name;
        if (config.isAutoRegisterTask()) {
            getZk().createIfNotExists(zkPath);
        }
        //判断是否分配给当前节点
        return getZk().exists(zkPath + "/" + uuid);
    }

}

class TimestampTypeAdapter implements JsonSerializer<Timestamp>, JsonDeserializer<Timestamp> {
    public JsonElement serialize(Timestamp src, Type arg1, JsonSerializationContext arg2) {
        DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String dateFormatAsString = format.format(new Date(src.getTime()));
        return new JsonPrimitive(dateFormatAsString);
    }

    public Timestamp deserialize(JsonElement json, Type typeOfT, JsonDeserializationContext context) throws JsonParseException {
        if (!(json instanceof JsonPrimitive)) {
            throw new JsonParseException("The date should be a string value");
        }
        try {
            DateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            Date date = format.parse(json.getAsString());
            return new Timestamp(date.getTime());
        } catch (Exception e) {
            throw new JsonParseException(e);
        }
    }
}